package cn.edu.gmall.realtime.app.dws;

import cn.edu.gmall.realtime.bean.VisitorStats;
import cn.edu.gmall.realtime.utils.ClickhouseUtil;
import cn.edu.gmall.realtime.utils.MyKafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);

        //检查点 CK 相关设置
//        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        StateBackend fsStateBackend = new FsStateBackend("hdfs://hadoop102:8020/gmall/flink/checkpoint/VisitorStatsApp");
//        env.setStateBackend(fsStateBackend);
//        System.setProperty("HADOOP_USER_NAME", "root");

        //TODO 1.从 Kafka 的 pv、uv、跳转明细主题中获取数据
        String pageViewSourceTopic = "dwd_page_log";
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";
        String groupId = "visitor_stats_app_6";

        FlinkKafkaConsumer<String> pageViewSource = MyKafkaUtil.getKafkaSource(pageViewSourceTopic, groupId);
        FlinkKafkaConsumer<String> uniqueVisitSource = MyKafkaUtil.getKafkaSource(uniqueVisitSourceTopic, groupId);
        FlinkKafkaConsumer<String> userJumpSource = MyKafkaUtil.getKafkaSource(userJumpDetailSourceTopic, groupId);

        DataStreamSource<String> pageViewDStream = env.addSource(pageViewSource);
        DataStreamSource<String> uniqueVisitDStream = env.addSource(uniqueVisitSource);
        DataStreamSource<String> userJumpDStream = env.addSource(userJumpSource);

        //TODO 2.对读取的流进行结构转换
        //2.1 转换 pv 流
        SingleOutputStreamOperator<VisitorStats> pageViewStatsDstream = pageViewDStream.map(
                json -> {
                    // System.out.println("pv:"+json);
                    JSONObject jsonObj = JSON.parseObject(json);
                    VisitorStats visitorStats = new VisitorStats(jsonObj);
                    visitorStats.setPv_ct(1L);
                    visitorStats.setDur_sum(jsonObj.getJSONObject("page").getLong("during_time"));
                    return visitorStats;
                });

        //2.2 转换 uv 流
        SingleOutputStreamOperator<VisitorStats> uniqueVisitStatsDstream = uniqueVisitDStream.map(
                json -> {
                    JSONObject jsonObj = JSON.parseObject(json);
                    VisitorStats visitorStats = new VisitorStats(jsonObj);
                    visitorStats.setUv_ct(1L);
                    return visitorStats;
                });

        //2.3 转换 sv 流
        SingleOutputStreamOperator<VisitorStats> sessionVisitDstream = pageViewDStream.process(
                new ProcessFunction<String, VisitorStats>() {
                    @Override
                    public void processElement(String json, ProcessFunction<String, VisitorStats>.Context ctx, Collector<VisitorStats> out) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(json);
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId == null || lastPageId.length() == 0) {
                            VisitorStats visitorStats = new VisitorStats(jsonObj);
                            visitorStats.setSv_ct(1L);
                            out.collect(visitorStats);
                        }
                    }
                });

        //2.4 转换跳转流
        SingleOutputStreamOperator<VisitorStats> userJumpStatDstream = userJumpDStream.map(
                json -> {
                    JSONObject jsonObj = JSON.parseObject(json);
                    VisitorStats visitorStats = new VisitorStats(jsonObj);
                    visitorStats.setUj_ct(1L);
                    return visitorStats;
                });

        //TODO 3.将四条流合并起来
        DataStream<VisitorStats> unionDetailDstream = uniqueVisitStatsDstream
                .union(pageViewStatsDstream, sessionVisitDstream, userJumpStatDstream);

        //TODO 4.设置水位线
        WatermarkStrategy<VisitorStats> visitorStatsWatermarkStrategy = WatermarkStrategy.
                <VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(1)).
                withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                    @Override
                    public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                        return element.getTs();
                    }
                });
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWatermarkDstream =
                unionDetailDstream.assignTimestampsAndWatermarks(visitorStatsWatermarkStrategy);

        //TODO 5.分组 选取四个维度作为 key , 使用 Tuple4 组合
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> visitorStatsTuple4KeyedStream = visitorStatsWithWatermarkDstream
                .keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                        return new Tuple4<>(visitorStats.getVc(), visitorStats.getCh(), visitorStats.getAr(), visitorStats.getIs_new());
                    }
                });

        //TODO 6.开窗
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowStream = visitorStatsTuple4KeyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 7.Reduce 聚合统计
        SingleOutputStreamOperator<VisitorStats> visitorStatsDstream = windowStream.reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats stats1, VisitorStats stats2) throws Exception {
                        //把度量数据两两相加
                        stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                        stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                        stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                        stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                        stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                        return stats1;
                    }
                },
                new ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                    @Override
                    public void process(Tuple4<String, String, String, String> Tuple4, Context context, Iterable<VisitorStats> visitorStatsIn, Collector<VisitorStats> visitorStatsOut) throws Exception {
                        // 补时间字段
                        for (VisitorStats visitorStats : visitorStatsIn) {
                            String startDate = simpleDateFormat.format(new Date(context.window().getStart()));
                            String endDate = simpleDateFormat.format(new Date(context.window().getEnd()));
                            visitorStats.setStt(startDate);
                            visitorStats.setEdt(endDate);
                            visitorStatsOut.collect(visitorStats);
                        }
                    }
                });

        visitorStatsDstream.print("reduce:");

        //TODO 8.向 ClickHouse 中写入数据
        String sql = "insert into visitor_stats values(?,?,?,?,?,?,?,?,?,?,?,?)";
        JdbcStatementBuilder<VisitorStats> jdbcStatementBuilder = new JdbcStatementBuilder<VisitorStats>() {
            @Override
            public void accept(PreparedStatement preparedStatement, VisitorStats visitorStats) throws SQLException {
                preparedStatement.setString(1, visitorStats.getStt());
                preparedStatement.setString(2, visitorStats.getEdt());
                preparedStatement.setString(3, visitorStats.getVc());
                preparedStatement.setString(4, visitorStats.getCh());
                preparedStatement.setString(5, visitorStats.getAr());
                preparedStatement.setString(6, visitorStats.getIs_new());
                preparedStatement.setLong(7, visitorStats.getUv_ct());
                preparedStatement.setLong(8, visitorStats.getPv_ct());
                preparedStatement.setLong(9, visitorStats.getSv_ct());
                preparedStatement.setLong(10, visitorStats.getUj_ct());
                preparedStatement.setLong(11, visitorStats.getDur_sum());
                preparedStatement.setLong(12, visitorStats.getTs());

            }
        };

        SinkFunction jdbcSink = ClickhouseUtil.getJdbcSink(sql, jdbcStatementBuilder);
        visitorStatsDstream.addSink(jdbcSink);

        // TODO 9.启动任务
        env.execute();

    }
}
